package com.example.service;

import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

@Component
@RocketMQMessageListener(topic = "", consumerGroup = "${rocketmq.consumer.group}")
public class RocketMQConsumerService implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    private static final Logger logger = LogManager.getLogger(RocketMQConsumerService.class);

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        try {
            // 订阅 default 和 leetcode 两个 topic
            consumer.subscribe("default", "*");
            consumer.subscribe("leetcode", "*");

            // 注册消息监听器
            consumer.registerMessageListener((MessageListenerConcurrently) (messages, context) -> {
                if (!CollectionUtils.isEmpty(messages)) {
                    for (MessageExt message : messages) {
                        String topic = message.getTopic();
                        String messageBody = new String(message.getBody());

                        logger.info("Received message: {} from topic: {}", messageBody, topic);

                        // 根据 topic 判断消息来自哪个主题，并处理相应逻辑
                        switch (topic) {
                            case "default":
                                handleDefault(topic, messageBody);
                                break;
                            case "leetcode":
                                handleLeetcode(topic, messageBody);
                                break;
                            default:
                                logger.warn("Received message from an unhandled topic: {}", topic);
                                break;
                        }
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
        } catch (MQClientException e) {
            logger.error("Error while setting up consumer", e);
        }
    }

    private void handleDefault(String topic, String message) {
        logger.info("Processing message from {} with content: {}", topic, message);
        // Default Topic 的处理逻辑
    }

    private void handleLeetcode(String topic, String message) {
        logger.info("Processing message from {} with content: {}", topic, message);
        // Leetcode Topic 的处理逻辑
    }

    @Override
    public void onMessage(String s) {

    }
}
